package thread

import (
	"fmt"
	"strings"
	"time"

	"gitee.com/sansaniot/ssiot-core/facade/env"
	"gitee.com/sansaniot/ssiot-core/facade/runtime"
	"gitee.com/sansaniot/ssiot-core/logger"

	"ssdevice/common/constant"
	"ssdevice/common/dto"
	"ssdevice/common/enums"
	"ssdevice/common/utils"
	"ssdevice/internal/device/models"
)

type DevHisDataThread struct {
}

var isPartition = false

func (DevHisDataThread) Run(threadName string) {
	logger.Infof("启动线程 - %s.", threadName)

	// 检测历史表类型(防止配置文件错误配置，以实际表信息为准)
	checkPart()

	// 初始化子线程个数
	constant.HisDataBucket = make([]*constant.CQueue, constant.DataRoutineChild)
	for i := 0; i < constant.DataRoutineChild; i++ {
		constant.HisDataBucket[i] = constant.NewCQueue()
		go func(base int) {
			child := fmt.Sprintf("%s-%d", threadName, base)
			logger.Infof("启动子线程 - %s", child)
			q := constant.HisDataBucket[base]
			dataList := make([]interface{}, 0)
			for {
				time.Sleep(1 * time.Second)
				for {
					s := q.Dequeue()
					if s == nil {
						break
					}
					dataList = append(dataList, s)
					if len(dataList) > 1000 {
						// 批量处理数据
						dealDevHistoryData(dataList, child)
						// 清空(减少扩容)
						dataList = dataList[0:0]
					}
				}
				if len(dataList) > 0 {
					// 批量处理数据
					dealDevHistoryData(dataList, child)
					// 清空(减少扩容)
					dataList = dataList[0:0]
				}
			}
		}(i)
	}
}

func dealDevHistoryData(dataList []interface{}, child string) {
	db := runtime.Runtime.GetDbByKey("db")
	if db == nil {
		return
	}
	sysHisSensorModel := &models.SysHistorySensor{}
	sysHisSensorModel.Orm = db
	sysHisSystemModel := &models.SysHistorySystem{}
	sysHisSystemModel.Orm = db
	sysDataDetailModel := &models.SysDataDetail{}
	sysDataDetailModel.Orm = db
	insertSensorList := make([]models.SysHistorySensor, 0)
	insertSystemList := make([]models.SysHistorySystem, 0)
	insertDTList := make([]models.SysDataDetail, 0)
	for _, v := range dataList {
		cacheData := utils.JsonStrToMap(v.(string))
		sn := cacheData["sn"].(string)
		updateTime, _ := utils.FormatStrTime(cacheData["updateTime"].(string))
		nowTime := dto.LocalTime(updateTime)
		dataMap := cacheData["data"].(map[string]interface{})
		msgType := cacheData["msgType"].(string)
		// 统计明细
		insertDTList = append(insertDTList, models.SysDataDetail{
			Type:       "DEV_DATA_SIZE",
			Key:        sn,
			Value:      len(utils.MapToBytes(dataMap)),
			CreateTime: nowTime,
		})
		for k, v := range dataMap {
			if msgType == enums.SensorMsg.Get().Name {
				insertSensorList = append(insertSensorList, models.SysHistorySensor{
					DevSn:      sn,
					Key:        k,
					Value:      utils.InterfaceToStr(v),
					UpdateTime: nowTime,
				})
				if isPartition {
					autoChildPartTable(sn, insertSensorList[0].TableName())
				}
			} else if msgType == enums.SystemMsg.Get().Name {
				insertSystemList = append(insertSystemList, models.SysHistorySystem{
					DevSn:      sn,
					Key:        k,
					Value:      utils.InterfaceToStr(v),
					UpdateTime: nowTime,
				})
			}
		}
	}

	if len(insertSensorList) > 0 {
		t1 := time.Now()
		if err := sysHisSensorModel.Add(&insertSensorList); err != nil {
			logger.Error(err)
		}
		logger.Infof("%s sensor历史记录批量入库%d. 耗时%v", child, len(insertSensorList), time.Since(t1).Round(time.Millisecond))
	}
	if len(insertSystemList) > 0 {
		logger.Infof("%s system历史记录入库%d.", child, len(insertSystemList))
		if err := sysHisSystemModel.Add(&insertSystemList); err != nil {
			logger.Error(err)
		}
	}
	if len(insertDTList) > 0 {
		if err := sysDataDetailModel.Add(&insertDTList); err != nil {
			logger.Error(err)
		}
	}
}

func checkPart() {
	scan := map[string]interface{}{}
	tableName := models.SysHistorySensor{}.TableName()
	if err := env.Db.Raw("SELECT relkind FROM pg_class WHERE relname = ?", tableName).Scan(&scan).Error; err != nil {
		logger.Error(err)
		return
	}
	if len(scan) == 0 {
		return
	}
	isPartition = scan["relkind"] == "p"
	if !isPartition {
		return
	}
	relnms := make([]map[string]interface{}, 0)
	sql0 := `SELECT relname FROM pg_class WHERE relnamespace = (SELECT relnamespace FROM pg_class where relname = '` + tableName + `') and relpartbound is not null`
	if err := env.Db.Raw(sql0).Scan(&relnms).Error; err != nil {
		logger.Error(err)
		return
	}
	for _, m := range relnms {
		constant.PartitionHistoryTbs = append(constant.PartitionHistoryTbs, utils.GetStringValueFiled(m, "relname"))
	}
}

func autoChildPartTable(sn, tableName string) {
	// 同建表的逻辑
	snn := strings.ReplaceAll(strings.ReplaceAll(strings.ToLower(sn), "-", "_"), " ", "_")
	partitionRelname := tableName + "_" + snn
	if utils.StrIn(partitionRelname, constant.PartitionHistoryTbs) {
		return
	}
	scan := map[string]interface{}{}
	if err := env.Db.Raw("create table if not EXISTS " + partitionRelname + " partition of " + tableName + " for values in ('" + sn + "') ").Scan(&scan).Error; err != nil {
		logger.Error(err)
		return
	}
	env.Db.Raw("drop index if EXISTS idx_" + partitionRelname + "_key_time").Scan(&scan)
	cidx := `CREATE INDEX "idx_` + partitionRelname + `_key_time" ON "` + partitionRelname + `" USING btree (
					"key" COLLATE "pg_catalog"."default" "pg_catalog"."text_ops" ASC NULLS LAST,
					"update_time" "pg_catalog"."timestamptz_ops" ASC NULLS LAST
				);`
	if err := env.Db.Raw(cidx).Scan(&scan).Error; err != nil {
		logger.Error(err)
		return
	}
	constant.PartitionHistoryTbs = append(constant.PartitionHistoryTbs, partitionRelname)
}
